package cn.topcodes.tcsf.app.service.impl;

import cn.topcode.unicorn.utils.JsonUtil;
import cn.topcode.unicorn.utils.StringUtil;
import cn.topcodes.tcsf.app.domain.push.PushProtocol;
import cn.topcodes.tcsf.app.service.PushService;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.ObjectPool;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.ddpush.im.v1.client.appserver.Pusher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.UnsupportedEncodingException;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * Created by Unicorn on 2017/7/26.
 */
@Service("pushService")
public class PushServiceImpl implements PushService {

    private Logger logger = LoggerFactory.getLogger(PushServiceImpl.class);

    private InternalPusher pusher = null;

    @Autowired
    private AppUserServiceImpl appUserService;

    @Autowired
    @Qualifier("asyncExecutor")
    private TaskExecutor taskExecutor;

    @Value("${ddpush.host}")
    private String ddpushHost;

    @Value("${ddpush.port}")
    private int ddpushPort;

    private int ddpushTimeout = 5000;

    @PostConstruct
    public void init() {
        try {
            pusher = new InternalPusher();
        } catch (Exception exception) {
            exception.printStackTrace();
        }
    }

    /**
     * 通过账号推送消息
     *
     * @param cmd     推送类型
     * @param content 推送内容
     */
    @Override
    public void pushByAccount(String account, String cmd, Object content) {
        List<String> tokens = appUserService.findTokensByUsername(account);
        if (tokens.isEmpty()) {
            return;
        }
        for (String token : tokens) {
            pushByToken(token, cmd, content);
        }
    }

    /**
     * 通过token推送消息
     *
     * @param token   用户登录凭证
     * @param cmd     推送类型
     * @param content 推送内容
     */
    @Override
    public void pushByToken(String token, String cmd, Object content) {
        PushProtocol pushProtocol = new PushProtocol();
        pushProtocol.setCmd(cmd);
        pushProtocol.setBody(JsonUtil.toJson(content));
        String json = JsonUtil.toJson(pushProtocol);
        this.pushByToken(token, json);
    }

    /**
     * 根据token推送文本
     *
     * @param token 用户登录凭证
     * @param text  文本
     */
    @Override
    public void pushByToken(String token, String text) {
        this.pushByToken(token, text.getBytes(Charset.forName("UTF-8")));
    }

    /**
     * 根据token推送二进制数据
     *
     * @param token 用户登录凭证
     * @param data  文本
     */
    @Override
    public void pushByToken(String token, byte[] data) {
        if (StringUtil.isBlank(token)) {
            logger.warn("推送token不能为空,token: " + token);
            return;
        }

        try {
            pusher.push0x20Message(md5(token), data);
        } catch (NoSuchAlgorithmException | UnsupportedEncodingException e) {
            throw new IllegalStateException(e);
        }
    }

    @PreDestroy
    public void destory() {
        pusher.close();
    }

    private byte[] md5(String text) throws NoSuchAlgorithmException, UnsupportedEncodingException {
        MessageDigest md = MessageDigest.getInstance("MD5");
        md.update(text.getBytes("UTF-8"));
        return md.digest();
    }

    class InternalPusher implements Runnable {

        private final ObjectPool<Pusher> internalPool;

        private final BlockingQueue<Message> queue = new LinkedBlockingQueue<>(100000);

        private volatile boolean stop;

        InternalPusher() throws Exception {
            GenericObjectPoolConfig config = new GenericObjectPoolConfig();
            config.setMaxWaitMillis(2000);
            internalPool = new GenericObjectPool<>(new PusherPooledObjectFactory(), config);
            new Thread(this).start();
        }

        class Message {
        }

        class PoisonPillMessage extends Message {
        }

        class InternalMessage extends Message {
            private final byte[] uuid;
            private final byte[] data;
            private int retries;

            InternalMessage(byte[] uuid, byte[] data) {
                this.uuid = uuid;
                this.data = data;
                this.retries = 0;
            }
        }

        void push0x20Message(byte[] uuid, byte[] data) {
            if (data == null) {
                throw new NullPointerException("data array is null");
            } else if (data.length >= 500) {
                throw new IllegalArgumentException("data array length illegal, min 1, max 500");
            } else {
                if (stop) {
                    throw new IllegalStateException("Pusher turned off");
                }
                if (!queue.offer(new InternalMessage(uuid, data))) {
                    logger.warn("推送消息队列已满");
                }
            }
        }

        @Override
        public void run() {
            Thread.currentThread().setName("internal-pusher-thread");
            try {
                while (true) {
                    Message msg = null;
                    try {
                        try {
                            // 避免队列中断直接结束推送线程
                            msg = queue.take();
                        } catch (InterruptedException ignored) {
                        }

                        if (msg instanceof PoisonPillMessage) {
                            // 关闭推送池
                            internalPool.close();
                            return;
                        } else if (msg instanceof InternalMessage) {
                            Message finalMsg = msg;
                            try {
                                taskExecutor.execute(() -> {
                                    InternalMessage message = (InternalMessage) finalMsg;
                                    Pusher pusher = null;
                                    try {
                                        try {
                                            pusher = internalPool.borrowObject();
                                        } catch (Exception ignored) {
                                        }

                                        if (pusher != null) {
                                            boolean result = pusher.push0x20Message(message.uuid, message.data);
                                            if (!result) {
                                                logger.warn("DDPUSH消息推送失败:" + new String(message.data, StandardCharsets.UTF_8));
                                            }
                                        }
                                    } catch (Exception exception) {
                                        // 忽略读写超时
                                        if (exception instanceof SocketTimeoutException) {
                                            exception.printStackTrace();
                                        }
                                        // data空指针入队列已经过滤
                                        else if (exception instanceof NullPointerException) {
                                            exception.printStackTrace();
                                        }
                                        // 推送长度超出上限入队列已经过滤
                                        else if (exception instanceof IllegalArgumentException) {
                                            exception.printStackTrace();
                                        }
                                        // 消息放入队列重发
                                        else {
                                            // SocketException直接关闭socket并且重新连接
                                            if (exception instanceof SocketException) {
                                                try {
                                                    internalPool.invalidateObject(pusher);
                                                } catch (Exception ignored) {
                                                }
                                                pusher = null;
                                            }
                                            logger.warn("DDPUSH消息推送失败:" + new String(message.data, StandardCharsets.UTF_8), exception);
                                            if (message.retries < 3) {
                                                message.retries++;
                                                logger.warn("开始重试发送消息:" + message.retries);
                                                queue.offer(message);
                                            }
                                        }
                                    } finally {
                                        // 重新放入推送池
                                        if (pusher != null) {
                                            try {
                                                internalPool.returnObject(pusher);
                                            } catch (Exception ignored) {
                                            }
                                        }
                                    }
                                });
                            } catch (Exception ex) {
                                // 线程池任务数量已满，抛出RejectedExecutionException，导致推送死循环退出
                                // 安卓App收不到推送，重新把推送消息放回队列

                                // 这个时候会有新的问题是之前没考虑到的，线程池不停拒绝任务并且抛出异常
                                // 本来这个时候线程池已经满了，如果被拒绝掉的任务没有限制的往线程池内加入任务会导致最新的消息推送不出去
                                // 这里只重试一次，这个时候本来就是高峰期，消息发送频繁；
                                // 重试多次也会影响最新的消息，所以宁愿丢弃部分消息也要保证推送可用
                                if (((InternalMessage) msg).retries < 1) {
                                    logger.warn("线程池任务上限，推送消息重新入列...");
                                    ((InternalMessage) msg).retries++;
                                    queue.offer(msg);
                                }
                            }
                        }
                    } catch (Exception ex) {
                        // 保险起见，再捕获一层
                        logger.warn("", ex);
                    }
                }
            } catch (Exception ignored) {
            }
        }

        void close() {
            stop = true;
            queue.offer(new PoisonPillMessage());
        }
    }

    class PusherPooledObjectFactory extends BasePooledObjectFactory<Pusher> {

        @Override
        public Pusher create() throws Exception {
            return new Pusher(ddpushHost, ddpushPort, ddpushTimeout);
        }

        @Override
        public PooledObject<Pusher> wrap(Pusher pusher) {
            return new DefaultPooledObject<>(pusher);
        }

        @Override
        public void destroyObject(PooledObject<Pusher> pooledObject) throws Exception {
            try {
                pooledObject.getObject().close();
            } catch (Exception ignored) {
            }
        }
    }
}
